fix(kafka): resolve disk-removal deadlock during rolling upgrade - #254

When a broker pod is deleted during rolling upgrade and a disk removal is
pending (GracefulDiskRemovalScheduled), the operator enters a deadlock:
reconcileKafkaPvc blocks the entire reconcile with "Disk removal pending",
preventing reconcileKafkaPod from recreating the missing pod. Meanwhile,
Cruise Control cannot complete the disk removal because the broker isn't
running.

Fix: move runningBrokers map building before reconcileKafkaPvc and pass it
in. Before returning the blocking error, check if any broker with pending
disk removal has a missing pod. If so, allow the reconcile to proceed so
the pod can be recreated. The disk removal check is re-evaluated on the
next cycle once the broker is back up.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

…pgrade

A terminally-failed or paused Cruise Control disk operation
(GracefulDisk{Removal,Rebalance}CompletedWithError / Paused) was treated
as "in progress" via IsDiskRemovalRunning(), so reconcileKafkaPvc returned
CruiseControlTaskRunning indefinitely. With all broker pods present this
froze the whole reconcile — including config rollout / rolling upgrade to
healthy brokers — since reconcileKafkaPod runs downstream of the PVC block.
Observed on a live 3-broker cluster stuck in ClusterRollingUpgrading with
every removed disk in GracefulDiskRemovalCompletedWithError.

Relaxing the block is data-safe: log.dirs retention
(shouldKeepRemovedLogDirInConfig) and PVC mount retention both keep the
removed disk in place until removal is confirmed *succeeded*, independent
of this block. Cruise Control also does not hang on a dead broker — it
marks the intra-broker task DEAD and completes with error — so a stalled
state is genuinely terminal, not in-flight work a restart could disrupt.

- add CruiseControlVolumeState.IsDiskOperationStalled() (CompletedWithError
  /Paused), mirroring IsDownscaleStalled
- in handleDiskRemoval, set waitForDiskRemovalToFinish only for non-stalled
  IsDiskRemoval()/IsDiskRebalance() states; branch selection (PVC deletion,
  state marking) is unchanged
- tests: removal CompletedWithError / Paused and rebalance CompletedWithError
  with pod present now proceed; Running + pod present still blocks

Complements the existing missing-pod bypass: narrowing handles a stalled
task with pods up; the bypass handles a genuinely-running task whose broker
pod is missing.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Add an e2e spec that applies a single manifest carrying both a read-only
broker config change (log.retention.hours, forces a rolling restart) and a
disk removal, then asserts the cluster reconciles correctly: the config
change propagates to broker ConfigMaps, the removed disk drops out of
log.dirs, Cruise Control goes quiescent, and the cluster returns to
ClusterRunning (a deadlock would time out this wait).

- config/samples/simplekafkacluster_1disk_configchange.yaml: 2disk sample
  reduced to one disk (removes /kafka-logs3) plus a changed readOnlyConfig
- tests/e2e/test_config_change_with_disk_removal.go: testConfigChangeWithDiskRemoval
  and brokerConfigMapsContainProperty helper
- wire the spec into the suite after testMultiDiskRemoval (chains 2->1 disk)

Covers the combined-operation happy path end to end; the stalled-removal and
missing-pod deadlocks the fix targets remain covered by the unit tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

The added stalled-state cases pushed the table-driven test past the funlen
limit (355 > 323). Annotate with //nolint:funlen, matching the existing
convention for long table tests in this file.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>